package net.pantolomin.nayo;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.atomic.AtomicBoolean;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The writer for data
 */
class DataWriter implements Runnable {
	/** The logger */
	private final Logger logger;

	/** The connection */
	private final NioConnection connection;

	/** The queue of data waiting to be written */
	private final Queue<ByteBuffer> queue;

	/** Authorizes only a single instance at a time */
	private final AtomicBoolean singleInstance;

	/**
	 * Create the writer
	 * 
	 * @param connection
	 *          the connection
	 */
	DataWriter(NioConnection connection) {
		this.logger = LoggerFactory.getLogger(DataWriter.class);
		this.connection = connection;
		this.queue = new ConcurrentLinkedQueue<ByteBuffer>();
		this.singleInstance = new AtomicBoolean(false);
	}

	/**
	 * Send some data on the channel
	 * 
	 * @param data
	 *          the data to send
	 */
	void addDataToSend(ByteBuffer data) {
		this.queue.add(data);
	}

	/**
	 * Ask to write on a channel
	 * 
	 * @param executor
	 *          the executor
	 * @throws RejectedExecutionException
	 *           if the writing can not be executed
	 */
	void askForWrite(ExecutorService executor) throws RejectedExecutionException {
		if (!this.singleInstance.getAndSet(true)) {
			askForExecution(executor);
		} // else: we are already writing ;o)
	}

	/**
	 * Resume writing once the write buffer is available
	 * 
	 * @param executor
	 *          the executor
	 * @throws RejectedExecutionException
	 *           if the writing can not be executed
	 */
	void resumeWriting(ExecutorService executor) throws RejectedExecutionException {
		boolean wasWriting = this.singleInstance.getAndSet(true);
		if (!wasWriting) {
			this.logger.warn("Writing resumed, but no writing was currently going");
		}
		try {
			this.connection.getConnector().getEventsListener().writeBufferFreed(this.connection);
		} catch (Exception e) {
			this.logger.warn("Failed to handle the event \"write buffer free\"", e);
		}
		askForExecution(executor);
	}

	/**
	 * Ask for the execution of the writing task
	 * 
	 * @param executor
	 *          the executor
	 * @throws RejectedExecutionException
	 *           if the writing can not be executed
	 */
	private void askForExecution(ExecutorService executor) throws RejectedExecutionException {
		try {
			executor.execute(this);
		} catch (RejectedExecutionException e) {
			// Can happen if someone tries to send a message on a connection
			// when the provider is stopped
			// Rmq: this also means that the connection is closed ;o)
			this.singleInstance.set(false);
			this.queue.clear();
			throw e;
		}
	}

	public void run() {
		try {
			// Write until there's no more data ...
			Iterator<ByteBuffer> itQueue = this.queue.iterator();
			if (itQueue.hasNext()) {
				// Prepare fairness check
				int fairness = this.connection.getProvider().getConfiguration().getWriteFairness();
				int iteration = 0;

				// Write loop
				do {
					ByteBuffer buf = itQueue.next();
					try {
						this.connection.getSocketChannel().write(buf);
					} catch (IOException e) {
						if (this.connection.isConnected()) {
							// An error although we are still connected
							try {
								this.connection.getConnector().getEventsListener().writeFailed(this.connection, e);
							} catch (Exception ex) {
								this.logger.warn("Failed to handle an error when writing", ex);
							}
						}
						this.singleInstance.set(false);
						return;
					}
					if (buf.remaining() > 0) {
						// ... or the socket's buffer fills up
						try {
							this.connection.getConnector().getEventsListener().writeBufferFull(this.connection);
						} catch (Exception e) {
							this.logger.warn("Failed to handle the event \"write buffer full\"", e);
						}
						// Ask for write once again (after a little moment)
						askForWrite();
						return;
					}
					itQueue.remove();
					if (!itQueue.hasNext()) {
						break;
					}

					// Check fairness
					iteration++;
					if (iteration >= fairness) {
						// Ask for write once again before getting out
						askForExecution(this.connection.getConnector().getProvider().getWriteExecutor());
						return;
					}
				} while (true);
			}
			this.singleInstance.set(false);
			itQueue = this.queue.iterator();
			if (itQueue.hasNext()) {
				// If a new element was added in the mean time, try to take the lock back
				askForWrite(this.connection.getConnector().getProvider().getWriteExecutor());
			}
		} catch (Throwable t) {
			this.singleInstance.set(false);
			this.logger.error("The writer failed its job", t);
		}
	}

	/**
	 * Indicate that we have data we want to write
	 */
	private void askForWrite() {
		this.connection.getProvider().getSelectionLoop().askForWrite(this.connection.getSelectionKey());
	}
}